/*
 * Copyright (c) 2003-2010, Live Media Pty. Ltd.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without modification, are permitted
 * provided that the following conditions are met:
 *
 *  1. Redistributions of source code must retain the above copyright notice, this list of
 *     conditions and the following disclaimer.
 *  2. Redistributions in binary form must reproduce the above copyright notice, this list of
 *     conditions and the following disclaimer in the documentation and/or other materials
 *     provided with the distribution.
 *  3. Neither the name of Live Media nor the names of its contributors may be used to endorse
 *     or promote products derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER
 * OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
 * ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package com.colony.channel;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import com.argot.TypeException;
import com.argot.TypeInputStream;
import com.argot.TypeLibrary;
import com.argot.TypeMap;
import com.argot.TypeMapperError;
import com.argot.TypeOutputStream;
import com.argot.common.UInt32;
import com.argot.remote.transport.TypeEndPoint;

import com.colony.concurrent.ThreadActivator;
import com.colony.transport.Client;

public class ChannelClient
{
	private static int U32_ID = 1;
	private static int CHANNEL_ID = 2;

	private Client _client;
	private TypeLibrary _library;
	private ThreadActivator _threadActivator;
	
	private TypeMap _channelTypeMap;
	
	public ChannelClient( Client client, TypeLibrary library, ThreadActivator threadActivator ) 
	throws TypeException
	{
		_client = client;
		_library = library;
		_threadActivator = threadActivator;
		
		_channelTypeMap = new TypeMap( _library, new TypeMapperError() );
		_channelTypeMap.map(U32_ID, library.getTypeId(UInt32.TYPENAME, UInt32.VERSION));
		_channelTypeMap.map(CHANNEL_ID, library.getTypeId(ChannelIdentifier.TYPENAME, ChannelIdentifier.VERSION));
		_channelTypeMap.map( 20, library.getDefinitionId( "meta.sequence", "1.3"));
		_channelTypeMap.map( 21, library.getDefinitionId( "meta.tag", "1.3"));
		_channelTypeMap.map( 22, library.getDefinitionId( "meta.reference", "1.3"));
		
	}
	
	private Channel connectChannel( int id )
	throws IOException, TypeException
	{
		TypeEndPoint end = _client.openLink();
		InputStream in = end.getInputStream();
		OutputStream out = end.getOutputStream();
		
		// send the request ID. 0 = assign.
		TypeOutputStream tout = new TypeOutputStream( out, _channelTypeMap );
		tout.writeObject(CHANNEL_ID, new ChannelIdentifier(id));
		tout.getStream().flush();
		
		// get the response ID. 0 = connected.
		TypeInputStream tin = new TypeInputStream( in, _channelTypeMap );
		ChannelIdentifier cid = (ChannelIdentifier) tin.readObject(CHANNEL_ID);
				
		Channel channel = new Channel( in, out, cid );
		return channel;
	}

	// client is attempting to send a input or output stream to the server.
	public ChannelIdentifier connectSendStream( InputStream in, OutputStream out ) 
	throws IOException, TypeException
	{
		Channel channel = connectChannel(0);
		if ( channel.getId().getId() == 0 )
		{
			throw new TypeException("Channel not assigned.");
		}
		
		if ( in != null )
		{
			_threadActivator.assignThread( new StreamReader(in, channel.getOutputStream()));
		}
		
		if ( out != null )
		{
			_threadActivator.assignThread( new StreamReader(channel.getInputStream(), out));
		}
		
		return channel.getId();
	}
	
	// client is receiving an input or output stream from the server.
	public Channel connectReceiveStream( ChannelIdentifier id ) 
	throws IOException, TypeException
	{
		return connectChannel(id.getId());
	}
	
	private class StreamReader
	implements Runnable
	{
		private InputStream _in;
		private OutputStream _out;
		
		public StreamReader( InputStream in, OutputStream out)
		{
			_in = in;
			_out = out;
		}
		
		public void run()
		{
			try
			{
				/*
				int x;
				System.out.println("StreamReader: reading bytes");
				while( (x = _in.read()) > 0 )
				{
					System.out.println("StreamReader: writing them");
					_out.write(x);
				}
				*/
				byte[] buffer = new byte[ 1024];
				int r;
				
				while ( ( r = _in.read(buffer)) > 0  )
				{
					System.out.println("Read " + r +" bytes.  Writing them out");
					_out.write( buffer, 0, r );
					_out.flush();
				}
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
			
			try
			{
				_in.close();
			}
			catch (IOException e)
			{ 
				e.printStackTrace();
			}
			
			try
			{
				_out.close();
			}
			catch (IOException e)
			{ 
				e.printStackTrace();
			}
			
		}
	}
}
